package com.dx.flink.event;

import com.dx.flink.serialize.FlinkCdcDataDeserializationSchema;
import com.dx.flink.sink.ImSink;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.Properties;

/**
 * 增量更新其它用户数据库用户到IM系统
 * 大致的流程就分为
 * 1.环境准备
 * 设置运行模式
 * 2.加载数据源
 * 3.数据转换
 * 4.数据输出
 * 5.执行程序
 *
 * @author xufeng
 * @since 07.04.2023
 */
@Component
public class OaUserEventListener implements ApplicationRunner {

    private final ImSink imSink;


    public OaUserEventListener(ImSink imSink) {
        this.imSink = imSink;
    }

    /**
     * 可以直接运行此类测试开发
     *
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        Properties debeziumProperties = new Properties();
        debeziumProperties.put("snapshot.locking.mode", "none");
        debeziumProperties.put("serverTimezone", "Asia/Shanghai");
        debeziumProperties.put("characterEncoding", "UTF-8");
        //2.加载数据源
        //创建 Flink-MySQL-CDC 的 Source
        //initial (default): 在第一次启动时对被监视的数据库表执行初始快照，并继续读取最新的binlog (开启断点续传后从上次消费offset继续消费)
        //latest-offset: 永远不要在第一次启动时对被监视的数据库表执行快照，只从binlog的末尾读取，这意味着只有自连接器启动以来的更改
        //timestamp: 永远不要在第一次启动时对监视的数据库表执行快照，直接从指定的时间戳读取binlog。使用者将从头遍历binlog，并忽略时间戳小于指定时间戳的更改事件
        //specific-offset: 不允许在第一次启动时对监视的数据库表进行快照，直接从指定的偏移量读取binlog。
        MySqlSource<String> build = MySqlSource.<String>builder()
                .serverTimeZone("Asia/Shanghai")
                .scanNewlyAddedTableEnabled(true) //开启支持新增表
                .hostname("192.168.2.253")
                .port(3306)
                .username("root")
                .password("123456")
                .databaseList("oa")
                //tableList为可选配置项,如果不指定该参数,则会读取上一个配置下的所有表的数据，注意：指定的时候需要使用"db.table"的方式.多表引文逗号隔开
                .tableList("oa.company_user")
                .startupOptions(StartupOptions.initial())
                //自定义反序列化器
                .deserializer(new FlinkCdcDataDeserializationSchema())
                .debeziumProperties(debeziumProperties)
                .build();


        // 启动一个webUI，指定本地WEB-UI端口号
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 18085);
        //检查点 改成参数传递进来
        //configuration.setString("execution.savepoint.path","file:///tmp/flink-ck/1980d53f557a886f885172bcdf4be8e8/chk-21");
        // 1.准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(configuration);
        env.setParallelism(1);
        //设置运行模式
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.enableCheckpointing(3000);
        // 设置本地
        env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-ck");

        // 5.执行程序
        //使用 CDC Source 从 MySQL 读取数据
        DataStreamSource<String> mysqlDS = env.fromSource(build, WatermarkStrategy.noWatermarks(), "MysqlSource");

        //打印数据
        mysqlDS.printToErr("------>").setParallelism(1);

        mysqlDS.addSink(new ImSink());

        //6.执行任务
        env.execute("OA_USER->IM_USER");
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        main(args.getSourceArgs());

    }


}
